package net.bwie.realtime.jtp.job;

import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.jtp.function.DimDataFilterFunction;
import net.bwie.realtime.jtp.function.DimHbaseSinkFunction;
import net.bwie.realtime.jtp.function.HbaseDimMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/06/02 16:31:15
 * @version: 1.0
 *  todo: 从Kafka队列topic-db实时获取维度表数据，将其同步存储HBase维表中，其中insert/update类型数据插入更新，delete类型数据删除
 *  todo: 第1. 多张维表；第2、多种类型数据
 */
public class RealtimeSyncDataDimJob {
    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        // 2. 数据源-source
//        DataStream<String> stream = KafkaUtil.consumerKafka(env, "topic-jtp-mall");
        DataStream<String> stream = KafkaUtil.consumerKafka(env, "topic-jtp-mall");
        //stream.print("kafka");

        // 3. 数据转换-transformation
        DataStream<String> dimStream = handle(stream);
//        dimStream.print("dim");

        // 4. 数据接收器-sink
        dimStream.addSink(new DimHbaseSinkFunction());

        // 5. 触发执行-execute
        env.execute("RealtimeAsyncDataDimJob");
    }

    private static DataStream<String> handle(DataStream<String> stream) {

        //todo: 维度表名称与主键映射

        HashMap<String, String> dimMap = new HashMap<String, String>() {{
            put("base_dic", "dic_code");
            put("base_province", "id");
            put("base_region", "id");
            put("base_category1", "id");
            put("base_category2", "id");
            put("base_category3", "id");
            put("base_trademark", "id");
            put("sku_info", "id");
            put("spu_info", "id");
            put("user_info", "id");
            put("activity_info", "id");
            put("activity_rule", "id");
            put("coupon_info", "id");
        }};

        // 维度表名称
        Set<String> dimSet = new HashSet<>(dimMap.keySet()) ;

        // s1-过滤维表数据
        SingleOutputStreamOperator<String> syncStream = stream.filter(new DimDataFilterFunction(dimSet));
//        syncStream.print("async");

        // s2-添加字段值
        SingleOutputStreamOperator<String> hbaseStream = syncStream.map(new HbaseDimMapFunction(dimMap));
//        hbaseStream.print("hbase");

        return hbaseStream;

    }
}
